package com.tvunetworks.center.user.redischannel;

import com.tvunetworks.center.user.service.UserDeviceService;
import com.tvunetworks.center.user.util.RedisUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.listener.PatternTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
import org.springframework.stereotype.Component;

/**
 * @program: tvucc-aggregator
 * @description:
 * @author lebronchen
 * @create: 2021-03-31 13:00
 **/
@Component
@Slf4j
public class UserRedisChannelListener {

    @Autowired
    private RedisUtil redisUtil;
    @Autowired
    private UserDeviceService userDeviceService;

    @Value("${showReceiverCentricRedisChannelLog:true}")
    private boolean showReceiverCentricRedisChannelLog;
    @Value("${receiverCentricRedisChannelTopic:CC_Device_Authorization}")
    private String receiverCentricRedisChannelTopic;

    class ReceiverCentricRedisChannelListener implements MessageListener {

        @Override
        public void onMessage(Message message, byte[] bytes) {
            byte[] body = message.getBody();
            try {
                String param = (String)redisUtil.deserialize(body);
                if (StringUtils.isBlank(param)) {
                    return;
                }
                if (param.startsWith("\"")) {
                    param = param.substring(1);
                }
                if (param.endsWith("\"")) {
                    param = param.substring(0, param.length() - 1);
                }

                if (showReceiverCentricRedisChannelLog) {
                    log.error("[ReceiverCentricRedisChannelListener] message:[{}]", param);
                }
                userDeviceService.notifyDevicePairChange(param);
            } catch (Exception e) {
                log.error(e.toString(), e);
            }
        }
    }

    /**
     *
     * @return MessageListenerAdapter
     */
    @Bean
    public MessageListenerAdapter receiverCentricRedisChannelListener() {
        return new MessageListenerAdapter(new ReceiverCentricRedisChannelListener());
    }

    /**
     *
     * @param connectionFactory connectionFactory
     * @return RedisMessageListenerContainer
     */
    @Bean
    public RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory) {
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        container.addMessageListener(receiverCentricRedisChannelListener(), new PatternTopic(receiverCentricRedisChannelTopic));
        return container;
    }

}
